package com.shujia.jinjie;

import com.shujia.utils.MyHBaseTools;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.ArrayList;

/*
    合理设计rowkey完成需求：
    1、通过手机号查询用户最近10条位置记录
    2、获取用户某一天在一个城市中的所有位置(组合过滤器)
==> tl_hefei_shushan_503.txt <==
D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C	117210031795040	83401	8340104	301	20180503190539	20180503233517	20180503
D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C	117205031830040	83401	8340104	510	20180503085547	20180503172154	20180503
D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C	117210031800040	83401	8340104	37	20180503180350	20180503180350	20180503
8827F3196977C6F752680505FEC0C7D3A18D4DFC	\N	\N	\N	\N	\N	\N	\N
 */
public class HBaseDianXinDemo {
    public static void main(String[] args) throws Exception {


        //获取数据库连接对象以及数据库操作对象
        //使用工具类获取
        ArrayList<Object> arrayList = MyHBaseTools.getConnection();
        Connection conn = (Connection) arrayList.get(0);
        Admin admin = (Admin) arrayList.get(1);

        //判断表是否存在
//        if (!MyHBaseTools.tableExists("dianxin_data", admin)) {
//            //使用工具类创建表（表名，列簇）
//            Boolean b = MyHBaseTools.createHbaseTable(admin, "dianxin_data", "info");
//            if (b) {
//                System.out.println("表建立成功！！");
//            } else {
//                System.out.println("表建立失败！！");
//            }
//        } else {
//            //调用方法添加数据
//            String path = "hbase/data/dianxin_data";
//            putsDianXianData(conn,path);
//        }

        //1、通过手机号查询用户最近10条位置记录
        ResultScanner resultScanner = scanData(conn, "dianxin_data");
        printData(resultScanner);


    }

    public static void printData(ResultScanner resultScanner){
        //遍历结果得到每一行
        for (Result result : resultScanner) {
            StringBuilder sb = new StringBuilder();
            //处理每一行
            String[] strings = Bytes.toString(result.getRow()).split("_");
            sb.append("手机号：");
            String phoneNum = strings[0];
            sb.append(phoneNum).append("，进入时间：");
            long enterTime = 20180504000000L - Long.parseLong(strings[1]);
            sb.append(enterTime).append("，网格编号：");
            //手机号,网格编号,城市编号,区县编号,停留时间,进入时间,离开时间,时间分区
            String wg = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("wg")));
            sb.append(wg).append("，城市编号：");
            String cityId = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("cityId")));
            sb.append(cityId).append("，区县编号：");
            String qx = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("qx")));
            sb.append(qx).append("，停留时间：");
            String stayTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("stayTime")));
            sb.append(stayTime).append("，离开时间：");
            String leaveTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("leaveTime")));
            sb.append(leaveTime).append("，时间分区：");
            String dateTime = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("dateTime")));
            sb.append(dateTime);

            System.out.println(sb);
        }
    }

    public static ResultScanner scanData(Connection conn,String tableName) throws Exception{
        //获取表的实例对象
        Table tableExample = MyHBaseTools.getTableExample(conn, tableName);

        Scan scan = new Scan();

        //创建行键前缀过滤器
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7"));
        scan.setFilter(prefixFilter);

        scan.setLimit(10);

        return tableExample.getScanner(scan);
    }


    //D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C	117210031795040	83401	8340104	301	20180503190539	20180503233517	20180503
    public static void putsDianXianData(Connection conn,String pathName) throws Exception {
        //创建字符缓冲输入流对象
        BufferedReader br = new BufferedReader(new FileReader(pathName));
        ArrayList<Put> puts = new ArrayList<>();
        String line = null;
        //遍历每一行数据，将每一行中的每一列封装成Put集合
        while ((line = br.readLine()) != null) {
            String[] strings = line.split("\t");
            if (strings.length > 7 && !"\\N".equals(strings[1])) {
                String phoneName = strings[0];
                String wg = strings[1];
                String cityId = strings[2];
                String qx = strings[3];
                String stayTime = strings[4];
                String enterTime = strings[5]; // 20180503190539
                String leaveTime = strings[6];
                String dateTime = strings[7];

                //设计rowkey,为了需求查询最近的10条记录做准备（时间戳反转）
                //将手机号与减之后的enterTime拼接
                //20180503190539  20180504000000
                long time = 20180504000000L - Long.parseLong(enterTime);
                String rowKey = phoneName + "_" + time;

                //将每一列封装成Put对象
                Put put = new Put(Bytes.toBytes(rowKey));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("wg"),Bytes.toBytes(wg));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("cityId"),Bytes.toBytes(cityId));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("qx"),Bytes.toBytes(qx));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("stayTime"),Bytes.toBytes(stayTime));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("leaveTime"),Bytes.toBytes(leaveTime));
                put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("dateTime"),Bytes.toBytes(dateTime));

                puts.add(put);
            }
        }

        //添加数据
        Table dianxinData = MyHBaseTools.getTableExample(conn, "dianxin_data");
        dianxinData.put(puts);
    }
}
